use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::sync::Arc;
use std::time::{Duration, Instant};
use tokio::sync::{Mutex, RwLock};
use tonic::transport::{Channel, Endpoint};
use tracing::{debug, error, info, instrument, warn};

/// RPC 服务配置
#[derive(Debug, Deserialize, Serialize, Clone)]
pub struct RpcConfig {
    /// System RPC 服务配置
    pub system: RpcServiceConfig,
    /// Infra RPC 服务配置  
    pub infra: InfraRpcConfig,
    /// 连接池配置
    pub pool: RpcPoolConfig,
}

/// RPC 服务连接配置
#[derive(Debug, Deserialize, Serialize, Clone)]
pub struct RpcServiceConfig {
    /// 服务地址
    pub address: String,
    /// 服务端口
    pub port: u16,
    /// 连接超时时间（秒）
    pub connect_timeout: u64,
    /// 请求超时时间（秒）
    pub request_timeout: u64,
    /// 是否启用
    pub enabled: bool,
}

/// Infra RPC 服务配置（包含多个子服务）
#[derive(Debug, Deserialize, Serialize, Clone)]
pub struct InfraRpcConfig {
    /// 字典服务配置
    pub dict: RpcServiceConfig,
    /// 通知服务配置
    pub notice: RpcServiceConfig,
    /// 文件服务配置
    pub file: RpcServiceConfig,
}

/// RPC 连接池配置
#[derive(Debug, Deserialize, Serialize, Clone)]
pub struct RpcPoolConfig {
    /// 最大连接数
    pub max_connections: usize,
    /// 最小空闲连接数
    pub min_idle: usize,
    /// 连接存活时间（秒）
    pub max_lifetime: u64,
    /// 空闲连接超时时间（秒）
    pub idle_timeout: u64,
}

impl Default for RpcConfig {
    fn default() -> Self {
        Self {
            system: RpcServiceConfig {
                address: "127.0.0.1".to_string(),
                port: 50050,
                connect_timeout: 30,
                request_timeout: 60,
                enabled: true,
            },
            infra: InfraRpcConfig {
                dict: RpcServiceConfig {
                    address: "127.0.0.1".to_string(),
                    port: 50052,
                    connect_timeout: 30,
                    request_timeout: 60,
                    enabled: true,
                },
                notice: RpcServiceConfig {
                    address: "127.0.0.1".to_string(),
                    port: 50053,
                    connect_timeout: 30,
                    request_timeout: 60,
                    enabled: true,
                },
                file: RpcServiceConfig {
                    address: "127.0.0.1".to_string(),
                    port: 50054,
                    connect_timeout: 30,
                    request_timeout: 60,
                    enabled: true,
                },
            },
            pool: RpcPoolConfig::default(),
        }
    }
}

impl Default for RpcPoolConfig {
    fn default() -> Self {
        Self {
            max_connections: 10,
            min_idle: 2,
            max_lifetime: 3600, // 1小时
            idle_timeout: 300,  // 5分钟
        }
    }
}

impl RpcServiceConfig {
    /// 获取完整的服务端点地址
    pub fn endpoint(&self) -> String {
        format!("http://{}:{}", self.address, self.port)
    }
}

/// RPC 连接信息
#[derive(Clone)]
pub struct RpcConnection {
    pub channel: Channel,
    created_at: Instant,
    last_used: Arc<Mutex<Instant>>,
}

impl RpcConnection {
    pub fn new(channel: Channel) -> Self {
        let now = Instant::now();
        Self {
            channel,
            created_at: now,
            last_used: Arc::new(Mutex::new(now)),
        }
    }

    pub async fn touch(&self) {
        *self.last_used.lock().await = Instant::now();
    }

    pub async fn is_expired(&self, max_lifetime: Duration, idle_timeout: Duration) -> bool {
        let now = Instant::now();
        let age = now.duration_since(self.created_at);
        let idle = now.duration_since(*self.last_used.lock().await);

        age > max_lifetime || idle > idle_timeout
    }
}

/// RPC 客户端连接池
#[derive(Clone)]
pub struct RpcConnectionPool {
    connections: Arc<RwLock<HashMap<String, Vec<RpcConnection>>>>,
    config: RpcConfig,
}

impl RpcConnectionPool {
    pub fn new(config: RpcConfig) -> Self {
        Self {
            connections: Arc::new(RwLock::new(HashMap::new())),
            config,
        }
    }

    /// 获取连接
    #[instrument(skip(self))]
    pub async fn get_connection(
        &self,
        service_name: &str,
        endpoint: &str,
    ) -> Result<Channel, RpcError> {
        // 尝试从连接池获取可用连接
        if let Some(conn) = self.try_get_from_pool(service_name).await {
            debug!("RPC连接池: 从池中获取 {} 连接", service_name);
            conn.touch().await;
            return Ok(conn.channel);
        }

        // 创建新连接
        debug!("RPC连接池: 为 {} 创建新连接", service_name);
        let channel = self.create_connection(endpoint).await?;

        // 将连接加入池中
        self.add_to_pool(service_name, channel.clone()).await;

        Ok(channel)
    }

    /// 从连接池获取连接
    async fn try_get_from_pool(&self, service_name: &str) -> Option<RpcConnection> {
        let mut connections = self.connections.write().await;

        if let Some(pool) = connections.get_mut(service_name) {
            // 清理过期连接
            let max_lifetime = Duration::from_secs(self.config.pool.max_lifetime);
            let idle_timeout = Duration::from_secs(self.config.pool.idle_timeout);

            pool.retain(|conn| {
                tokio::task::block_in_place(|| {
                    tokio::runtime::Handle::current()
                        .block_on(async { !conn.is_expired(max_lifetime, idle_timeout).await })
                })
            });

            // 返回可用连接
            pool.pop()
        } else {
            None
        }
    }

    /// 创建新连接
    async fn create_connection(&self, endpoint: &str) -> Result<Channel, RpcError> {
        let endpoint = Endpoint::from_shared(endpoint.to_string())
            .map_err(|e| RpcError::InvalidEndpoint(e.to_string()))?
            .connect_timeout(Duration::from_secs(30))
            .timeout(Duration::from_secs(60));

        let channel = endpoint
            .connect()
            .await
            .map_err(|e| RpcError::ConnectionFailed(e.to_string()))?;
        Ok(channel)
    }

    /// 添加连接到池中
    async fn add_to_pool(&self, service_name: &str, channel: Channel) {
        let mut connections = self.connections.write().await;
        let pool = connections
            .entry(service_name.to_string())
            .or_insert_with(Vec::new);

        // 检查连接池大小限制
        if pool.len() < self.config.pool.max_connections {
            pool.push(RpcConnection::new(channel));
        }
    }

    /// 清理过期连接
    #[instrument(skip(self))]
    pub async fn cleanup_expired(&self) {
        let mut connections = self.connections.write().await;
        let max_lifetime = Duration::from_secs(self.config.pool.max_lifetime);
        let idle_timeout = Duration::from_secs(self.config.pool.idle_timeout);

        for (service_name, pool) in connections.iter_mut() {
            let before_count = pool.len();
            pool.retain(|conn| {
                tokio::task::block_in_place(|| {
                    tokio::runtime::Handle::current()
                        .block_on(async { !conn.is_expired(max_lifetime, idle_timeout).await })
                })
            });
            let after_count = pool.len();

            if before_count != after_count {
                debug!(
                    "RPC连接池: {} 清理了 {} 个过期连接",
                    service_name,
                    before_count - after_count
                );
            }
        }
    }
}

/// RPC 客户端错误类型
#[derive(Debug, thiserror::Error)]
pub enum RpcError {
    #[error("服务 {0} 已禁用")]
    ServiceDisabled(String),

    #[error("连接失败: {0}")]
    ConnectionFailed(String),

    #[error("无效的端点: {0}")]
    InvalidEndpoint(String),

    #[error("RPC调用失败: {code:?} - {message}")]
    RpcCallFailed { code: tonic::Code, message: String },

    #[error("请求参数无效: {0}")]
    InvalidArgument(String),

    #[error("资源未找到: {0}")]
    NotFound(String),

    #[error("服务不可用: {0}")]
    Unavailable(String),

    #[error("内部服务器错误: {0}")]
    Internal(String),
}

impl From<tonic::Status> for RpcError {
    fn from(status: tonic::Status) -> Self {
        match status.code() {
            tonic::Code::InvalidArgument => Self::InvalidArgument(status.message().to_string()),
            tonic::Code::NotFound => Self::NotFound(status.message().to_string()),
            tonic::Code::Unavailable => Self::Unavailable(status.message().to_string()),
            _ => Self::RpcCallFailed {
                code: status.code(),
                message: status.message().to_string(),
            },
        }
    }
}

// Salvo StatusError 转换
impl From<RpcError> for salvo::prelude::StatusError {
    fn from(error: RpcError) -> Self {
        match error {
            RpcError::InvalidArgument(msg) => {
                salvo::prelude::StatusError::bad_request().brief(&msg)
            }
            RpcError::NotFound(msg) => salvo::prelude::StatusError::not_found().brief(&msg),
            RpcError::ServiceDisabled(msg) | RpcError::Unavailable(msg) => {
                salvo::prelude::StatusError::service_unavailable().brief(&msg)
            }
            _ => salvo::prelude::StatusError::internal_server_error().brief(&error.to_string()),
        }
    }
}

/// AppError 转换为 tonic::Status 的辅助函数
pub fn convert_app_error_to_status(err: crate::response::AppError) -> tonic::Status {
    match err {
        crate::response::AppError::NotFound { message, .. } => tonic::Status::not_found(message),
        crate::response::AppError::Validation { message, .. } => {
            tonic::Status::invalid_argument(message)
        }
        crate::response::AppError::Database { message, .. } => tonic::Status::internal(message),
        crate::response::AppError::Unauthorized { .. } => {
            tonic::Status::permission_denied("权限不足")
        }
        crate::response::AppError::Internal { message, .. } => tonic::Status::internal(message),
        crate::response::AppError::ExternalService { message, .. } => {
            tonic::Status::unavailable(message)
        }
        crate::response::AppError::Business { message, .. } => {
            tonic::Status::failed_precondition(message)
        }
        crate::response::AppError::Configuration { message, .. } => {
            tonic::Status::internal(format!("配置错误: {}", message))
        }
        crate::response::AppError::Network { message, .. } => {
            tonic::Status::unavailable(format!("网络错误: {}", message))
        }
        crate::response::AppError::Timeout { message, .. } => {
            tonic::Status::deadline_exceeded(format!("超时错误: {}", message))
        }
        crate::response::AppError::Rpc { message, .. } => {
            tonic::Status::internal(format!("RPC错误: {}", message))
        }
        crate::response::AppError::Grpc { message, grpc_code, .. } => {
            tonic::Status::new(grpc_code, format!("gRPC错误: {}", message))
        }
    }
}

/// 多服务 RPC 客户端管理器
#[derive(Clone)]
pub struct RpcClientManager {
    pool: RpcConnectionPool,
    config: RpcConfig,
}

impl RpcClientManager {
    /// 创建 RPC 客户端管理器
    #[instrument]
    pub fn new(config: RpcConfig) -> Self {
        info!("RPC客户端管理器: 初始化，支持服务: system, infra-dict, infra-notice, infra-file");

        let manager = Self {
            pool: RpcConnectionPool::new(config.clone()),
            config,
        };

        // 启动定期清理任务
        let pool_for_cleanup = manager.pool.clone();
        tokio::spawn(async move {
            let mut interval = tokio::time::interval(Duration::from_secs(60));
            loop {
                interval.tick().await;
                pool_for_cleanup.cleanup_expired().await;
            }
        });

        manager
    }

    /// 获取指定服务的连接
    pub async fn get_connection(&self, service_name: &str) -> Result<Channel, RpcError> {
        let endpoint = match service_name {
            "system" => {
                if !self.config.system.enabled {
                    return Err(RpcError::ServiceDisabled("system".to_string()));
                }
                self.config.system.endpoint()
            }
            "dict" => {
                if !self.config.infra.dict.enabled {
                    return Err(RpcError::ServiceDisabled("dict".to_string()));
                }
                self.config.infra.dict.endpoint()
            }
            "notice" => {
                if !self.config.infra.notice.enabled {
                    return Err(RpcError::ServiceDisabled("notice".to_string()));
                }
                self.config.infra.notice.endpoint()
            }
            "file" => {
                if !self.config.infra.file.enabled {
                    return Err(RpcError::ServiceDisabled("file".to_string()));
                }
                self.config.infra.file.endpoint()
            }
            _ => return Err(RpcError::ServiceDisabled(service_name.to_string())),
        };

        self.pool.get_connection(service_name, &endpoint).await
    }

    /// 健康检查
    #[instrument(skip(self))]
    pub async fn health_check(&self) -> HashMap<String, bool> {
        let mut results = HashMap::new();

        // 检查各个服务
        for service in &["system", "dict", "notice", "file"] {
            match self.get_connection(service).await {
                Ok(_) => {
                    results.insert(service.to_string(), true);
                }
                Err(e) => {
                    warn!("RPC健康检查: {} 服务不可用: {}", service, e);
                    results.insert(service.to_string(), false);
                }
            }
        }

        results
    }
}

/// RPC 服务器辅助函数
pub mod server {
    use std::net::SocketAddr;
    use tonic::transport::Server;
    use tracing::info;

    /// 启动 RPC 服务器的通用逻辑
    ///
    /// 注意：这是一个简化的辅助函数，用于统一服务器启动逻辑
    /// 具体的服务需要在各自的应用中实现
    pub async fn start_rpc_server_with_addr(
        addr: SocketAddr,
        service_name: &str,
    ) -> Result<Server, Box<dyn std::error::Error>> {
        info!("🚀 {} RPC Server 准备启动...", service_name);
        info!("🌐 RPC 服务器监听地址: {}", addr);

        Ok(Server::builder())
    }

    /// 记录服务器启动信息
    pub fn log_server_startup(service_name: &str, addr: &SocketAddr) {
        info!("🚀 {} RPC Server 启动中...", service_name);
        info!("🌐 RPC 服务器监听地址: {}", addr);
    }
}

/// RPC调用辅助工具
pub mod helpers {
    use super::*;
    use salvo::prelude::*;
    use crate::response::*;

    /// 处理RPC调用并转换为Salvo响应
    pub async fn handle_rpc_call<T, R, F>(
        rpc_call: F,
        success_message: Option<&str>,
        mapper: impl FnOnce(R) -> T,
    ) -> Result<Json<ApiResponse<T>>, StatusError>
    where
        F: std::future::Future<Output = Result<tonic::Response<R>, tonic::Status>>,
        T: serde::Serialize,
    {
        match rpc_call.await {
            Ok(response) => {
                info!("API: RPC 调用成功");
                let rpc_res = response.into_inner();
                let data = mapper(rpc_res);
                
                match success_message {
                    Some(msg) => Ok(Json(ApiResponse::success_with_message(data, msg))),
                    None => Ok(Json(ApiResponse::success(data))),
                }
            }
            Err(status) => {
                error!("API: RPC 调用失败: {:?}", status);
                Err(convert_rpc_status_to_salvo_error(status))
            }
        }
    }

    /// 将tonic::Status转换为Salvo StatusError
    pub fn convert_rpc_status_to_salvo_error(status: tonic::Status) -> StatusError {
        match status.code() {
            tonic::Code::Ok => {
                StatusError::internal_server_error().brief("意外的成功状态码")
            }
            tonic::Code::Cancelled => {
                StatusError::service_unavailable().brief("请求已取消")
            }
            tonic::Code::Unknown => {
                StatusError::internal_server_error().brief("未知错误")
            }
            tonic::Code::InvalidArgument => {
                StatusError::bad_request().brief("请求参数无效")
            }
            tonic::Code::DeadlineExceeded => {
                StatusError::request_timeout().brief("请求超时")
            }
            tonic::Code::NotFound => {
                StatusError::not_found().brief("资源未找到")
            }
            tonic::Code::AlreadyExists => {
                StatusError::conflict().brief("资源已存在")
            }
            tonic::Code::PermissionDenied => {
                StatusError::forbidden().brief("权限不足")
            }
            tonic::Code::ResourceExhausted => {
                StatusError::service_unavailable().brief("资源已耗尽")
            }
            tonic::Code::FailedPrecondition => {
                StatusError::unprocessable_entity().brief(&format!("前置条件失败: {}", status.message()))
            }
            tonic::Code::Aborted => {
                StatusError::conflict().brief("操作已中止")
            }
            tonic::Code::OutOfRange => {
                StatusError::bad_request().brief("参数超出范围")
            }
            tonic::Code::Unimplemented => {
                StatusError::not_implemented().brief("功能未实现")
            }
            tonic::Code::Internal => {
                StatusError::internal_server_error().brief("服务器内部错误")
            }
            tonic::Code::Unavailable => {
                StatusError::service_unavailable().brief("服务暂时不可用")
            }
            tonic::Code::DataLoss => {
                StatusError::internal_server_error().brief("数据丢失")
            }
            tonic::Code::Unauthenticated => {
                StatusError::unauthorized().brief("认证失败")
            }
        }
    }

    /// 常用的RPC错误处理宏
    #[macro_export]
    macro_rules! handle_rpc {
        ($rpc_call:expr, $mapper:expr) => {
            $crate::rpc::helpers::handle_rpc_call($rpc_call, None, $mapper).await
        };
        ($rpc_call:expr, $mapper:expr, $message:expr) => {
            $crate::rpc::helpers::handle_rpc_call($rpc_call, Some($message), $mapper).await
        };
    }

    /// 简单的RPC调用宏（当响应类型相同时）
    #[macro_export]
    macro_rules! simple_rpc {
        ($rpc_call:expr) => {
            $crate::rpc::helpers::handle_rpc_call($rpc_call, None, |x| x).await
        };
        ($rpc_call:expr, $message:expr) => {
            $crate::rpc::helpers::handle_rpc_call($rpc_call, Some($message), |x| x).await
        };
    }
}
